package com.itcam.kafka.listener;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.support.ProducerListener;
import org.springframework.stereotype.Component;

/**
 * @author : Cammy.Wu
 * Description : 
 */

@Component
@Slf4j
public class KafkaSendResultHandler implements ProducerListener {

    /**
     * 当 Kafka 消息发送成功时的回调方法。
     *
     * @param producerRecord 包含发送到 Kafka 的消息的生产者记录，其中包含了消息的主题、键、值等信息。
     * @param recordMetadata 包含消息在 Kafka 中的元数据，如分区、偏移量等。
     */
    @Override
    public void onSuccess(ProducerRecord producerRecord, RecordMetadata recordMetadata) {
        // 从生产者记录中获取消息的主题
        String topic = producerRecord.topic();
        // 从生产者记录中获取消息的值，并将其转换为字符串
        String value = producerRecord.value().toString();
        // 从记录元数据中获取消息所在的分区
        Integer partition = recordMetadata.partition();
        // 使用日志记录器记录消息发送成功的信息，包含主题、值和分区
        log.info("topic:{},value:{},partition:{},发送成功回调", topic, value, partition);
    }


    /**
     * 当 Kafka 消息发送失败时的回调方法。
     *
     * @param producerRecord 包含发送到 Kafka 的消息的生产者记录，其中包含了消息的主题、键、值等信息。
     * @param recordMetadata 包含消息在 Kafka 中的元数据，如分区、偏移量等。
     * @param exception 发送消息时抛出的异常，包含了失败的原因。
     */
    @Override
    public void onError(ProducerRecord producerRecord, RecordMetadata recordMetadata, Exception exception) {
        // 从生产者记录中获取消息的主题
        String topic = producerRecord.topic();
        // 从生产者记录中获取消息的值，并将其转换为字符串
        String value = producerRecord.value().toString();
        // 使用日志记录器记录消息发送失败的信息，包含主题、值和失败原因
        log.info("topic:{},value:{},发送失败的原因:{}", topic, value, exception.getMessage());
    }

}
